package org.raftkv.raft.confchange;

import org.raftkv.commons.Tuple;
import org.raftkv.raft.exceptions.RaftException;
import org.raftkv.raft.pb.ConfChangeSingle;
import org.raftkv.raft.quorum.JointConfig;
import org.raftkv.raft.quorum.MajorityConfig;
import org.raftkv.raft.tracker.Config;
import org.raftkv.raft.tracker.Inflights;
import org.raftkv.raft.tracker.Progress;
import org.raftkv.raft.tracker.ProgressMap;
import org.raftkv.raft.tracker.ProgressTracker;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

public class Changer {

    private ProgressTracker tracker;

    private long lastIndex;

    private static final Logger LOGGER = LoggerFactory.getLogger(Changer.class);

    public Changer(ProgressTracker tracker, long lastIndex) {
        this.tracker = tracker;
        this.lastIndex = lastIndex;
    }

    public Tuple<Config, ProgressMap> enterJoint(boolean autoLeave, List<ConfChangeSingle> ccs) {
        Tuple<Config, ProgressMap> t = checkAndCopy();
        Config cfg = t.getFirst();
        ProgressMap prs = t.getSecond();

        if (joint(cfg)) {
            throw new RaftException("config is already joint");
        }

        if (incoming(cfg.getVoters()).size() == 0) {
            throw new RaftException("can't make a zero-voter config joint");
        }

        cfg.getVoters().CONFIGS[1] = new MajorityConfig();

        for (Long id : incoming(cfg.getVoters())) {
            outgoing(cfg.getVoters()).remove(id);
        }

        apply(cfg, prs, ccs);
        cfg.setAutoLeave(autoLeave);
        return checkAndReturn(cfg, prs);
    }

    public Tuple<Config, ProgressMap> leaveJoint() {
        Tuple<Config, ProgressMap> t = checkAndCopy();
        Config cfg = t.getFirst();
        ProgressMap prs = t.getSecond();

        if (!joint(cfg)) {
            throw new RaftException("can't leave a non-joint config");
        }

        if (outgoing(cfg.getVoters()).size() == 0) {
            throw new RaftException(String.format("configuration is not joint: %s", cfg));
        }

        for (Long id : cfg.getLearnersNext()) {
            Set<Long> m = new HashSet<>();
            if (cfg.getLearners() == null) {
                cfg.setLearners(m);
            }
            m.add(id);
            prs.getProgressMap().get(id).setLearner(true);
        }
        cfg.setLearnersNext(null);

        for (Long id : outgoing(cfg.getVoters())) {
            boolean isVoter = incoming(cfg.getVoters()).contains(id);
            boolean isLearner = cfg.getLearners().contains(id);

            if (!isVoter && !isLearner) {
                prs.getProgressMap().remove(id);
            }
        }

        cfg.getVoters().CONFIGS[1] = null;
        cfg.setAutoLeave(false);

        return checkAndReturn(cfg, prs);
    }

    public Tuple<Config, ProgressMap> simple(List<ConfChangeSingle> ccs) {
        Tuple<Config, ProgressMap> t = checkAndCopy();
        Config cfg = t.getFirst();
        ProgressMap prs = t.getSecond();

        if (joint(cfg)) {
            throw new RaftException("can't apply simple config change in joint config");
        }

        apply(cfg, prs, ccs);
        int n = symdiff(incoming(tracker.getConfig().getVoters()), incoming(cfg.getVoters()));
        if (n > 1) {
            throw new RaftException("more than one voter changed without entering joint config");
        }

        return checkAndReturn(cfg, prs);
    }

    private void apply(Config cfg, ProgressMap prs, List<ConfChangeSingle> ccs) {
        for (ConfChangeSingle cc : ccs) {
            if (cc.getNodeId() == 0) {
                continue;
            }

            switch (cc.getType()) {
                case ConfChangeAddNode:
                    makeVoter(cfg, prs, cc.getNodeId());
                    break;
                case ConfChangeAddLearnerNode:
                    makeLearner(cfg, prs, cc.getNodeId());
                    break;
                case ConfChangeRemoveNode:
                    remove(cfg, prs, cc.getNodeId());
                    break;
                case ConfChangeUpdateNode:
                    break;
                default:
                    String errMsg = String.format("unexpected conf type %d", cc.getType().ordinal());
                    LOGGER.error(errMsg);
                    throw new RaftException(errMsg);
            }
        }
        if (incoming(cfg.getVoters()).size() == 0) {
            throw new RaftException("removed all voters");
        }
    }

    private void makeVoter(Config cfg, ProgressMap prs, long id) {
        Progress pr = prs.getProgressMap().get(id);
        if (pr == null) {
            initProgress(cfg, prs, id, false);
            return;
        }

        pr.setLearner(false);
        nilAwareDelete(cfg, cfg.getLearners(), id);
        nilAwareDelete(cfg, cfg.getLearnersNext(), id);
        incoming(cfg.getVoters()).add(id);
    }

    private void makeLearner(Config cfg, ProgressMap prs, long id) {
        Progress pr = prs.getProgressMap().get(id);
        if (pr == null) {
            initProgress(cfg, prs, id, true);
            return;
        }

        if (pr.isLearner()) {
            return;
        }

        remove(cfg, prs, id);
        prs.getProgressMap().put(id, pr);

        if (outgoing(cfg.getVoters()).contains(id)) {
            Set<Long> m = new HashSet<>();
            if (cfg.getLearnersNext() == null) {
                cfg.setLearnersNext(m);
            }
            m.add(id);
        } else {
            pr.setLearner(true);
            Set<Long> m = new HashSet<>();
            if (cfg.getLearners() == null) {
                cfg.setLearners(m);
            }
            m.add(id);
        }
    }

    private void remove(Config cfg, ProgressMap prs, long id) {
        Progress pr = prs.getProgressMap().get(id);
        if (pr == null) {
            return;
        }

        incoming(cfg.getVoters()).remove(id);
        nilAwareDelete(cfg, cfg.getLearners(), id);
        nilAwareDelete(cfg, cfg.getLearnersNext(), id);

        if (outgoing(cfg.getVoters()) != null && !outgoing(cfg.getVoters()).contains(id)) {
            prs.getProgressMap().remove(id);
        }
    }

    private void initProgress(Config cfg, ProgressMap prs, long id, boolean isLearner) {
        if (!isLearner) {
            incoming(cfg.getVoters()).add(id);
        } else {
            Set<Long> m = new HashSet<>();
            if (cfg.getLearners() == null) {
                cfg.setLearners(m);
            }
            m.add(id);
        }

        Progress t = new Progress();
        t.setNext(lastIndex);
        t.setMatch(0);
        t.setInflights(Inflights.newInflights(tracker.getMaxInflight()));
        t.setLearner(isLearner);
        t.setRecentActive(true);
        prs.getProgressMap().put(id, t);
    }

    private static void checkInvariants(Config cfg, ProgressMap prs) {
        List<Set<Long>> t = new ArrayList<>();
        t.add(cfg.getVoters().ids());
        t.add(cfg.getLearners());
        t.add(cfg.getLearnersNext());

        for (Set<Long> ids : t) {
            if (ids != null) {
                for (Long id : ids) {
                    if (!prs.getProgressMap().containsKey(id)) {
                        String errMsg = String.format("no progress for %d", id);
                        LOGGER.error(errMsg);
                        throw new RaftException(errMsg);
                    }
                }
            }
        }

        if (cfg.getLearnersNext() != null) {
            for (Long id : cfg.getLearnersNext()) {
                if (outgoing(cfg.getVoters()) == null || !outgoing(cfg.getVoters()).contains(id)) {
                    String errMsg = String.format("%d is in LearnersNext, but not Voters[1]", id);
                    LOGGER.error(errMsg);
                    throw new RaftException(errMsg);
                }

                if (prs.getProgressMap().get(id).isLearner()) {
                    String errMsg = String.format("%d is in LearnersNext, but is already marked as learner", id);
                    LOGGER.error(errMsg);
                    throw new RaftException(errMsg);
                }
            }
        }

        if (cfg.getLearners() != null) {
            for (Long id : cfg.getLearners()) {
                if (outgoing(cfg.getVoters()) != null && outgoing(cfg.getVoters()).contains(id)) {
                    String errMsg = String.format("%d is in Learners and Voters[1]", id);
                    LOGGER.error(errMsg);
                    throw new RaftException(errMsg);
                }

                if (incoming(cfg.getVoters()).contains(id)) {
                    String errMsg = String.format("%d is in Learners and Voters[0]", id);
                    LOGGER.error(errMsg);
                    throw new RaftException(errMsg);
                }

                if (!prs.getProgressMap().get(id).isLearner()) {
                    String errMsg = String.format("%d is in Learners, but is not marked as learner", id);
                    LOGGER.error(errMsg);
                    throw new RaftException(errMsg);
                }
            }
        }

        if (!joint(cfg)) {
            if (outgoing(cfg.getVoters()) != null) {
                String errMsg = "cfg.Voters[1] must be nil when not joint";
                LOGGER.error(errMsg);
                throw new RaftException(errMsg);
            }

            if (cfg.getLearnersNext() != null) {
                String errMsg = "cfg.LearnersNext must be nil when not joint";
                LOGGER.error(errMsg);
                throw new RaftException(errMsg);
            }

            if (cfg.isAutoLeave()) {
                String errMsg = "AutoLeave must be false when not joint";
                LOGGER.error(errMsg);
                throw new RaftException(errMsg);
            }
        }
    }

    private Tuple<Config, ProgressMap> checkAndCopy() {
        Config cfg = tracker.getConfig().clone();
        ProgressMap prs = new ProgressMap();

        for (Map.Entry<Long, Progress> entry : tracker.getProgress().getProgressMap().entrySet()) {
            Long id = entry.getKey();
            Progress pr = entry.getValue();

            prs.getProgressMap().put(id, pr);
        }
        return checkAndReturn(cfg, prs);
    }

    private static Tuple<Config, ProgressMap> checkAndReturn(Config cfg, ProgressMap prs) {
        try {
            checkInvariants(cfg, prs);
        } catch (Exception e) {
            return new Tuple<>(new Config(), new ProgressMap());
        }
        return new Tuple<>(cfg, prs);
    }

    private static void nilAwareDelete(Config cfg, Set<Long> m, long id) {
        if (m == null) {
            return;
        }
        m.remove(id);
        if (m.size() == 0) {
            if (m == cfg.getLearners()) {
                cfg.setLearners(null);
            }
            if (m == cfg.getLearnersNext()) {
                cfg.setLearnersNext(null);
            }
        }
    }

    private static int symdiff(MajorityConfig l, MajorityConfig r) {
        int n = 0;
        List<List<MajorityConfig>> pairs = new ArrayList<>();
        pairs.add(List.of(l, r));
        pairs.add(List.of(r, l));

        for (List<MajorityConfig> p : pairs) {
            for (Long id : p.get(0)) {
                if (!p.get(1).contains(id)) {
                    n++;
                }
            }
        }
        return n;
    }

    private static boolean joint(Config cfg) {
        return outgoing(cfg.getVoters()) != null && outgoing(cfg.getVoters()).size() > 0;
    }

    private static MajorityConfig incoming(JointConfig voters) {
        return voters.CONFIGS[0];
    }

    private static MajorityConfig outgoing(JointConfig voters) {
        return voters.CONFIGS[1];
    }

    public static String describe(List<ConfChangeSingle> ccs) {
        StringBuilder buf = new StringBuilder();
        for (ConfChangeSingle cc : ccs) {
            if (buf.length() > 0) {
                buf.append(' ');
            }
            buf.append(String.format("%s(%d)", cc.getType(), cc.getNodeId()));
        }
        return buf.toString();
    }

    public ProgressTracker getTracker() {
        return tracker;
    }

    public void setTracker(ProgressTracker tracker) {
        this.tracker = tracker;
    }

    public long getLastIndex() {
        return lastIndex;
    }

    public void setLastIndex(long lastIndex) {
        this.lastIndex = lastIndex;
    }

}
